package service

import (
	"context"
	"gitee.com/crack007/goose"
	"github.com/Shopify/sarama"
	"github.com/crack007/go-im/src/common/config"
)

type KafkaConsumer struct {
	consumer          sarama.Consumer
	ctx               context.Context
	partitionConsumer sarama.PartitionConsumer
}

func NewKafkaConsumer(ctx context.Context) *KafkaConsumer {
	return &KafkaConsumer{ctx: ctx}
}
func (k *KafkaConsumer) Init() error {
	go func() {
		<-k.ctx.Done()
		k.Close()
	}()
	var err error
	k.consumer, err = sarama.NewConsumer(config.KafkaConfig.Host, nil)
	if err == nil {
		goose.GetLogger().Debug("Kafka connected")
	}
	return err
}
func (k *KafkaConsumer) Consumer(topic string, msgCallback func(msg *sarama.ConsumerMessage)) (err error) {
	goose.GetLogger().Debug("Consumer topic->%s", topic)
	k.partitionConsumer, err = k.consumer.ConsumePartition(topic, 0, sarama.OffsetNewest)
	if err != nil {
		return err
	}
	for msg := range k.partitionConsumer.Messages() {
		msgCallback(msg)
	}
	return nil
}

func (k *KafkaConsumer) Close() {
	if k.partitionConsumer != nil {
		k.partitionConsumer.Close()
	}
	if k.consumer != nil {
		k.consumer.Close()
	}
}
